/*
 * Copyright 1999-2018 Alibaba Group Holding Ltd.
 *
 * Licensed under the Apache License, Version 2.0 (the "License");
 * you may not use this file except in compliance with the License.
 * You may obtain a copy of the License at
 *
 *      http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 */

package com.alibaba.nacos.naming.core.v2.service.impl;

import com.alibaba.nacos.api.exception.NacosException;
import com.alibaba.nacos.api.exception.runtime.NacosRuntimeException;
import com.alibaba.nacos.api.naming.pojo.Instance;
import com.alibaba.nacos.api.naming.utils.NamingUtils;
import com.alibaba.nacos.common.notify.NotifyCenter;
import com.alibaba.nacos.naming.core.v2.ServiceManager;
import com.alibaba.nacos.naming.core.v2.client.Client;
import com.alibaba.nacos.naming.core.v2.client.manager.ClientManager;
import com.alibaba.nacos.naming.core.v2.client.manager.ClientManagerDelegate;
import com.alibaba.nacos.naming.core.v2.event.client.ClientOperationEvent;
import com.alibaba.nacos.naming.core.v2.event.metadata.MetadataEvent;
import com.alibaba.nacos.naming.core.v2.pojo.BatchInstancePublishInfo;
import com.alibaba.nacos.naming.core.v2.pojo.InstancePublishInfo;
import com.alibaba.nacos.naming.core.v2.pojo.Service;
import com.alibaba.nacos.naming.core.v2.service.ClientOperationService;
import com.alibaba.nacos.naming.misc.Loggers;
import com.alibaba.nacos.naming.pojo.Subscriber;
import org.springframework.stereotype.Component;

import java.util.ArrayList;
import java.util.List;

/**
 * Operation service for ephemeral clients and services.
 *
 * @author xiweng.yy
 */
@Component("ephemeralClientOperationService")
public class EphemeralClientOperationServiceImpl implements ClientOperationService {
    
    private final ClientManager clientManager;
    
    public EphemeralClientOperationServiceImpl(ClientManagerDelegate clientManager) {
        this.clientManager = clientManager;
    }
    
    @Override
    public void registerInstance(Service service, Instance instance, String clientId) throws NacosException {
        NamingUtils.checkInstanceIsLegal(instance);

        //获取服务管理对象
        ServiceManager serviceManager = ServiceManager.getInstance();

        /**
         * 从服务管理对象，获取服务对象.
         */
        Service singleton = serviceManager.getSingleton(service);

        //判断是否是临时实例，使用grpc方式，一定是临时实例的，如果是持久化那么就会报错
        if (!singleton.isEphemeral()) {
            throw new NacosRuntimeException(NacosException.INVALID_PARAM,
                    String.format("Current service %s is persistent service, can't register ephemeral instance.",
                            singleton.getGroupedServiceName()));
        }



        /**
         * 获取一个  Client 对象，Client从哪里来的？它在服务注册实例注册之前会添加一个 Client 对象。
         *
         */
        Client client = clientManager.getClient(clientId);
        if (!clientIsLegal(client, clientId)) {
            return;
        }

        //这个步骤就是把请求过来instance对象里面的属性复制一份到 InstancePublishInfo 对象上

        InstancePublishInfo instanceInfo = getPublishInfo(instance);

        /**
         * 添加服务实例到 Client 对象中存储（核心重点步骤）：
         *  1. 里面代码逻辑可以看出，就是把服务实例对象添加到map里
         *      key：Service 对象
         *      value：InstancePublishInfo 服务实例信息对象
         *
         * 问题：
         *  1. 好像和我想的不一样，一个服务下有多个服务实例 value 存储的理应存储的是
         *       key：Service 对象，value：List<InstancePublishInfo> 服务实例信息对象；
         *      怎么会是 key：Service 对象，value：InstancePublishInfo 服务实例信息对象？
         *      难到这个不是 nacos server端的服务注册表？其实这个就是 nacos server 端的服务注册表。
         *
         *
         *
         *  2. 接下来要梳理对象之间的关系，这点是非常重要：
         *           clientManager：是管理证每个客户端对象
         *             Client：对象代表与客户端建立的连接和通信，你每个项目注册进来都会创建一个 Client 对象。场景如：现在是我有一个订单服务，但是有两个订单服务实例对象。那么 Client 和  订单服务 以及 订单服务实例对象 对象是什么关系呢？
         *
         *              nacos server：
         *                  order1 向 nacos server 注册服务，此时 nacos server会创建 Client 对象，该对象是代表于 order1 之间的连接通信，然后创建一个 Service 对象 以及 ServiceInstance。最后把这两个对象存储到map里
         *                      key：Service
         *                      value： serviceInstance
         *                  order2 向 nacos server 注册服务，此时 nacos server 也会创建 Client 对象。该对象是代表于 order2之间的连接通信，然后创建一个 Service 对象 以及 ServiceInstance。最后把这两个对象存储到map里
         *                      key：Service
         *                      value： serviceInstance
         *
         *                  这个 Service 理论上是 同属一样的。
         *
         *
         *             通过上面的分析可以发现：
         *                1. 如果是按 nacos 代码角度来分析：一个服务下有两个服务实例对象。那么它是创建两个 Client 对象，每一个 Client 对象会存储 一个 Service 对象 以及 ServiceInstance。
         *                    如果获取服务实例对象时候，那么肯定是先得到 Client 对象然后从 Client 获取服务实例对象，获取同一个服务下的多个实例，那么就会遍历多个 Client 对象
         *                    这个就是 nacos server 对服务实例的存储结构方式，这点要理解清楚！！！
         *
         *                2. 如果按照理论逻辑来分析：我们认为的是 nacos server 它理应是这样子存储，它会创建一个 Service 对象，该 Service 对象里面维护着多个 ServiceInstance 实例对象。
         *                    这种存储结构其实是存在并发操作问题的，这点要清楚
         *                   如果获取服务实例对象时候：就是定位到时哪个服务对象，然后从该服务对象获取服务实例。
         *                   这个就是我们通常的认知思维 对服务实例的存储结构方式。
         *
         *               总结：我们通常认为的存储方式实际和 nacos server 存储方式 是不同的，这样一定要记清楚！！！！！
         *
         *
         *  3. 现在已经知道了 nacos server 对于 serviceInstance 服务实例的存储是在 Client 对象上。那么我想获取该服务下的多个实例。我怎么知道 哪个 Client 对象是属于同一个 Service 服务下的？
         *     所以我们可以猜测，肯定会有一个地方存储着 Service 以及多个 Client 之间的关系。代码逻辑就在下面！！！就在 发布 ClientRegisterServiceEvent 事件逻辑上！！！！
         *       发布 ClientRegisterServiceEvent 事件，该事件就是走两件事：
         *          1. 存储着 Service 以及多个 Client 之间的关系。存储结果肯定是一个 Map，
         *               key： Service 对象
         *               value： Set<String> 集合，该集合存储的是 clientId 字符串
         *           2. 在发布一个服务变动事件
         *
         *     最后总结：通过 Service 对象 获取多个 clientId ，再通过 clientId 从 clientManager 获取 Client 对象，然后获取 ServiceInstance 服务实例对象信息，最后返回！！！
         */
        client.addServiceInstance(singleton, instanceInfo);
        //设置更新时间
        client.setLastUpdatedTime();

        client.recalculateRevision();

        /**
         *
         * 发布 ClientRegisterServiceEvent 事件，该事件就是走两件事（核心） ：
         * 1. 存储着 Service 以及多个 Client 之间的关系。这个就是 nacos server 端的服务注册表 ，存储结果肯定是一个 Map，
         *     key： Service 对象
         *     value： Set<String> 集合，该集合存储的是 clientId 字符串
         *
         * 2. 在发布一个 ServiceChangedEvent 服务变动事件，该事件通知其它订阅者，该服务发生了改变。你要可能要更新下你本地的服务信息。如：
         *      现在有A 、B服务两个服务。A要调用B服务，A向 nacos server获取 B服务的信息，然后把 B 服务信息存储到本地上。A调用B服务都是从本地获取的。那么如果B服务发生了改变。
         *      A服务的怎么更新自己的本地B服务信息缓存？这个事件就是通知A服务你订阅的B服务发送了改变了你要更新下自己本地缓存
         *
         */
        NotifyCenter.publishEvent(new ClientOperationEvent.ClientRegisterServiceEvent(singleton, clientId));
        NotifyCenter.publishEvent(new MetadataEvent.InstanceMetadataEvent(singleton, instanceInfo.getMetadataId(), false));
    }
    
    @Override
    public void batchRegisterInstance(Service service, List<Instance> instances, String clientId) {
        Service singleton = ServiceManager.getInstance().getSingleton(service);
        if (!singleton.isEphemeral()) {
            throw new NacosRuntimeException(NacosException.INVALID_PARAM,
                    String.format("Current service %s is persistent service, can't batch register ephemeral instance.",
                            singleton.getGroupedServiceName()));
        }
        Client client = clientManager.getClient(clientId);
        if (!clientIsLegal(client, clientId)) {
            return;
        }
        BatchInstancePublishInfo batchInstancePublishInfo = new BatchInstancePublishInfo();
        List<InstancePublishInfo> resultList = new ArrayList<>();
        for (Instance instance : instances) {
            InstancePublishInfo instanceInfo = getPublishInfo(instance);
            resultList.add(instanceInfo);
        }
        batchInstancePublishInfo.setInstancePublishInfos(resultList);
        client.addServiceInstance(singleton, batchInstancePublishInfo);
        client.setLastUpdatedTime();
        NotifyCenter.publishEvent(new ClientOperationEvent.ClientRegisterServiceEvent(singleton, clientId));
        NotifyCenter.publishEvent(
                new MetadataEvent.InstanceMetadataEvent(singleton, batchInstancePublishInfo.getMetadataId(), false));
    }
    
    @Override
    public void deregisterInstance(Service service, Instance instance, String clientId) {
        if (!ServiceManager.getInstance().containSingleton(service)) {
            Loggers.SRV_LOG.warn("remove instance from non-exist service: {}", service);
            return;
        }
        Service singleton = ServiceManager.getInstance().getSingleton(service);
        Client client = clientManager.getClient(clientId);
        if (!clientIsLegal(client, clientId)) {
            return;
        }
        InstancePublishInfo removedInstance = client.removeServiceInstance(singleton);
        client.setLastUpdatedTime();
        client.recalculateRevision();
        if (null != removedInstance) {
            NotifyCenter.publishEvent(new ClientOperationEvent.ClientDeregisterServiceEvent(singleton, clientId));
            NotifyCenter.publishEvent(
                    new MetadataEvent.InstanceMetadataEvent(singleton, removedInstance.getMetadataId(), true));
        }
    }
    
    @Override
    public void subscribeService(Service service, Subscriber subscriber, String clientId) {

        Service singleton = ServiceManager.getInstance().getSingletonIfExist(service).orElse(service);

        Client client = clientManager.getClient(clientId);
        if (!clientIsLegal(client, clientId)) {
            return;
        }
        // 添加服务订阅者
        client.addServiceSubscriber(singleton, subscriber);
        client.setLastUpdatedTime();
        //发布客户端订阅服务事件
        NotifyCenter.publishEvent(new ClientOperationEvent.ClientSubscribeServiceEvent(singleton, clientId));
    }
    
    @Override
    public void unsubscribeService(Service service, Subscriber subscriber, String clientId) {
        Service singleton = ServiceManager.getInstance().getSingletonIfExist(service).orElse(service);
        Client client = clientManager.getClient(clientId);
        if (!clientIsLegal(client, clientId)) {
            return;
        }
        client.removeServiceSubscriber(singleton);
        client.setLastUpdatedTime();
        NotifyCenter.publishEvent(new ClientOperationEvent.ClientUnsubscribeServiceEvent(singleton, clientId));
    }
    
    private boolean clientIsLegal(Client client, String clientId) {
        if (client == null) {
            Loggers.SRV_LOG.warn("Client connection {} already disconnect", clientId);
            return false;
        }
        if (!client.isEphemeral()) {
            Loggers.SRV_LOG.warn("Client connection {} type is not ephemeral", clientId);
            return false;
        }
        return true;
    }
}
